import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object AdImpClick {
  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val fileClick = "F:\\lagou\\lagouhomework\\stage_4_module_2\\3.广告统计曝光数和点击数\\AdImpClick\\data\\click.log"
    val fileImp = "F:\\lagou\\lagouhomework\\stage_4_module_2\\3.广告统计曝光数和点击数\\AdImpClick\\data\\imp.log"
    val fileOut = "F:\\lagou\\lagouhomework\\stage_4_module_2\\3.广告统计曝光数和点击数\\AdImpClick\\data\\output.dat"

    val clicklines: RDD[String] = sc.textFile(fileClick)
    //    clicklines.foreach(println)
    /** 18005474为没曝光记录的测试数据
     * INFO 2019-09-01 00:31:51 requestURI:/click?app=1&p=1&adid=18005472&industry=469
     * INFO 2019-09-01 00:31:51 requestURI:/click?app=1&p=1&adid=18005474&industry=469
     * INFO 2019-09-01 00:29:53 requestURI:/click?app=1&p=1&adid=18005472&industry=469
     * INFO 2019-09-01 00:30:31 requestURI:/click?app=2&p=1&adid=18005472&industry=469
     * INFO 2019-09-01 00:31:03 requestURI:/click?app=1&p=1&adid=18005472&industry=469
     */

    val clickData = clicklines.map { line =>
      val fields = line.split('&')
      val kv = fields(2).split('=')
      (kv(1), 1)
    }
    //    clickData.foreach(println)
    /**
     * (18005472,1)
     * (18005472,1)
     * (18005472,1)
     * (18005472,1)
     * (18005474,1)
     */

    val sumClick = clickData.reduceByKey(_ + _).sortBy(_._1, false)
    //    sumClick.foreach(println)
    /**
     * (18005474,1)
     * (18005472,4)
     */
    val implines: RDD[String] = sc.textFile(fileImp)
    //    implines.foreach(println)

    /** 18005473为没有点击记录的测试数据
     * INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469
     * INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469
     * INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005472&industry=469
     * INFO 2019-09-01 00:29:53 requestURI:/imp?app=1&p=1&adid=18005473&industry=469
     */

    val impData = implines.map { line =>
      val fields = line.split('&')
      val kv = fields(2).split('=')
      (kv(1), 1)
    }
    //    impData.foreach(println)

    /**
     * (18005472,1)
     * (18005473,1)
     * (18005472,1)
     * (18005472,1)
     */


    val sumImp = impData.reduceByKey(_ + _).sortBy(_._1)
    //    sumImp.foreach(println)
    /**
     * (18005473,1)
     * (18005472,3)
     */

    val mergeData = sumClick.cogroup(sumImp)
    //    mergeData.foreach(println)
    /**
     * (18005473,(CompactBuffer(),CompactBuffer(1)))
     * (18005472,(CompactBuffer(4),CompactBuffer(3)))
     * (18005474,(CompactBuffer(1),CompactBuffer()))
     */
    val outData = mergeData.map {
      case (str, tuple) =>
        val clickNum = tuple._1.sum
        val impNum = tuple._2.sum

        (str, clickNum, impNum)
    }.sortBy(_._1)

    outData.foreach(println)
    outData.saveAsTextFile(fileOut)

    /**
     * ad_id      click   imp
     * (18005472, 4,      3)
     * (18005473, 0,      1)    //只曝光没点击
     * (18005474, 1,      0)    //只点击没曝光
     */
    // 关闭SparkContext
    sc.stop()
  }
}
